package org.etlt.job;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.etlt.EtltRuntimeException;
import org.etlt.SettingReader;
import org.etlt.check.CheckList;
import org.etlt.check.ChecklistFactory;
import org.etlt.expression.ExpressionEvaluator;
import org.etlt.expression.VariableContext;
import org.etlt.expression.datameta.Variable;
import org.etlt.expression.function.FunctionFactory;
import org.etlt.extract.Entity;
import org.etlt.extract.Extractor;
import org.etlt.extract.ExtractorFactory;
import org.etlt.job.config.xml.JobSettingXmlSimplerResolver;
import org.etlt.load.Loader;
import org.etlt.load.LoaderFactory;
import org.etlt.load.LoaderStatus;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * this class is not thread safe, when it is used in many threads, copy it firstly
 * <p>Usage:</p>
 * 1)JobContext context = new JobContext();<br>
 * 2)context.init() {@link #init()}<br>
 */
public class JobContext implements VariableContext {

    protected Log log = LogFactory.getLog(getClass());

    private static final Map<String, SettingResolver<JobSetting>> JOB_SETTING_RESOLVERS = new HashMap<>();

    static {
//        JOB_SETTING_RESOLVERS.put("xml", new JobSettingXmlResolver());
        JOB_SETTING_RESOLVERS.put("xml", new JobSettingXmlSimplerResolver());
    }

    public static final String JOB_RECORD = "job-record.json";

    private static final String JOB_SETTING_XML = "job.xml";


    private final ExpressionEvaluator expressionEvaluator = new ExpressionEvaluator();
    /**
     *
     */
    private Map<String, Entity> entityContainer = new HashMap<>();

    private final Map<String, Object> resourceContainer = new ConcurrentHashMap<>();

    private File contextRoot;

    private RunMode runMode = RunMode.RENEW;

    private JobSetting jobSetting;

    private SettingResolver<JobSetting> settingResolver;

    private final SettingReader reader = new SettingReader();

    private JobMonitor jobMonitor = new JobMonitor(this);

    private final Map<String, Extractor> extractors = new HashMap<>();

    private final Map<String, Loader> loaders = new HashMap<>();
    /**
     * execution finished loaders
     */
    private Map<String, Loader> finishedLoaders = new HashMap<>();
    /**
     * the loaders which don't depend on any loaders
     */
    private final Map<String, Loader> independentLoaders = new HashMap<>();

    private final Map<String, CheckList> checklists = new HashMap<>();

    public JobExecution getJobExecution() {
        return jobExecution;
    }

    private JobExecution jobExecution = new JobExecution(this);

    private Map<String, Object> mapping = new HashMap<>();

    public JobContext(File contextRoot) throws IOException {
        if (contextRoot.isDirectory()) {
            this.contextRoot = contextRoot;
            resolveSetting();
        } else {
            this.contextRoot = contextRoot.getParentFile();
            resolveSetting(contextRoot);
        }
    }

    private JobContext() {

    }

    public ExpressionEvaluator getExpressionEvaluator() {
        return expressionEvaluator;
    }

    public RunMode getRunMode() {
        return runMode;
    }

    public void setRunMode(RunMode runMode) {
        this.runMode = runMode;
    }

    public JobContext copy() {
        JobContext copiedContext = new JobContext();
        copiedContext.contextRoot = this.contextRoot;
        copiedContext.jobSetting = this.jobSetting;
        copiedContext.extractors.putAll(this.extractors);
        copiedContext.loaders.putAll(this.loaders);
        copiedContext.finishedLoaders = this.finishedLoaders;

        copiedContext.checklists.putAll(this.checklists);
        copiedContext.mapping.putAll(this.mapping);
        copiedContext.resourceContainer.putAll(this.resourceContainer);
        copiedContext.jobExecution = this.jobExecution;
        copiedContext.jobExecution.setContext(copiedContext);
        copiedContext.jobMonitor = this.jobMonitor;
        /**
         * entity container should be initialized
         */
        copiedContext.entityContainer = new HashMap<>();
        return copiedContext;
    }

    private SettingResolver<JobSetting> findSettingResolver(File setting) {
        String type = getSettingType(setting);
        SettingResolver<JobSetting> settingResolver = JOB_SETTING_RESOLVERS.get(type);
        if (settingResolver == null)
            throw new IllegalStateException("setting resolver not found for " + setting.getName());
        return settingResolver;
    }

    private String getSettingType(File settingFile) {
        String name = settingFile.getName();
        int index = name.indexOf(".");
        return index == -1 ? name : name.substring(index + 1);
    }

    /**
     * choose a job setting resolver, job.xml is preferred, job.json is secondly preferred
     */
    protected void resolveSetting() throws FileNotFoundException {
        File jobXml = new File(this.contextRoot, JOB_SETTING_XML);
        if (jobXml.exists()) {
            resolveSetting(jobXml);
        }  else
            throw new IllegalStateException("missing job setting file, job.json and job.xml are missing, one of both is needed.");
    }

    /**
     * choose a job setting resolver, job.xml is preferred, job.json is secondly preferred
     */
    protected void resolveSetting(File jobFile) throws FileNotFoundException {
        if (jobFile.exists()) {
            settingResolver = findSettingResolver(jobFile);
            this.jobSetting = settingResolver.resolve(new FileInputStream(jobFile), this);
            this.jobSetting.getFunctionTypes().forEach(type -> {
                getExpressionEvaluator().registerFunctions(type);
            });
        } else
            throw new IllegalStateException("missing job setting file, only .xml are supported: " + jobFile);
    }

    public JobSetting getJobSetting() {
        return jobSetting;
    }

    public File getContextRoot() {
        return contextRoot;
    }

    public boolean exists(String entity) {
        return this.entityContainer.containsKey(entity);
    }

    public Object getValue(String entity, String column) {
        if (exists(entity)) {
            return this.entityContainer.get(entity).getValue(column);
        }
        return null;
    }

    public Entity getEntity(String entityName) {
        if (exists(entityName)) {
            return this.entityContainer.get(entityName);
        }
        return null;
    }

    public String getPropertyValue(String name) {
        return getJobSetting().getPropertyValue(name);
    }

    public Map<String, Object> getMapping(){
        return this.mapping;
    }
    /**
     * @param catalog catalog
     * @param key     key
     * @return mapping result
     */
    public Object mapping(String catalog, String key) {
        if (StringUtils.isBlank(catalog))
            throw new IllegalArgumentException("catalog name of mapping cannot be blank.");
        if (StringUtils.isBlank(key))
            return key;
        Map cmap = ((Map) this.mapping.get(catalog));
        if (Objects.isNull(cmap))
            throw new IllegalArgumentException("catalog of mapping missing, check mapping configuration.");
        return cmap.get(key.trim());
    }

    public void setEntity(String entityName, Entity entity) {
        this.entityContainer.put(entityName, entity);
    }

    public void removeEntity(String entity) {
        this.entityContainer.remove(entity);
    }

    /**
     * read a job setting {@link JobSetting} from job.json file located in {@link #contextRoot}
     */
    public void init() throws IOException {
        log.info("init resources ...");
        initResources();
        /*
          init all extractors
         */
        log.info("init extractors ...");
        initExtractors();
//        initBundleExtractors();
        /*
         * init all loaders
         */
        log.info("init loaders ...");
        initLoaders();
//        initBundleLoaders();
        /*
         * init all validators
         */
        log.info("init checklists ...");
        initChecklists();
        /*
         * init mapping
         */
        log.info("init mapping resources ...");
        initMapping();
        /*
         * start job monitor
         */
        this.jobMonitor.start();
    }

    private void addExtractor(Extractor extractor) {
        if (this.extractors.containsKey(extractor.getName()))
            throw new EtltRuntimeException("duplicated extractor found: " + extractor.getName());
        List<String> extractorNames = this.jobSetting.getExtractors();
        if (!extractorNames.contains(extractor.getName()))
            this.extractors.put(extractor.getName(), extractor);
    }

    private void addLoader(Loader loader) {
        if (this.loaders.containsKey(loader.getName()))
            throw new EtltRuntimeException("duplicated loader found: " + loader.getName());
        List<String> loaderNames = this.jobSetting.getLoaders();
        if (!loaderNames.contains(loader.getName()))
            this.loaders.put(loader.getName(), loader);
    }


    /**
     *
     * @throws IOException
     */
    protected void initExtractors() {
        this.getJobSetting().getExtractors().forEach((resourceName) -> {
            try {
                List<Extractor> extractors =
                        ExtractorFactory.getInstance().resolveInstances(resourceName, JobContext.this);
                extractors.forEach((extractor -> {
                    addExtractor(extractor);
                }));
            } catch (IOException e) {
                throw new IllegalStateException(e);
            }
        });
    }

    /**
     *
     * @throws IOException
     */
    protected void initLoaders() throws IOException {
        getJobSetting().getLoaders().forEach((resourceName) -> {
            try {
                List<Loader> loaders = LoaderFactory.getInstance().resolveInstance(resourceName, JobContext.this);
                loaders.forEach((loader -> {
                    addLoader(loader);
                }));
            } catch (IOException e) {
                throw new IllegalStateException(e);
            }
        });

        checkLoaders();

        reorgLoaders();

        if (getRunMode() == RunMode.RESTORE) {
            this.jobExecution.restore();
            List<JobExecution.ExecutionItem> items = this.jobExecution.getExecutionItems();
            items.forEach(item -> {
                String player = item.getPlayer();
                Loader loader = this.loaders.get(player);
                if (loader != null && (LoaderStatus.valueOf(item.getStatus()) == LoaderStatus.SUCCESS)) {
                    this.finishedLoaders.put(loader.getName(), loader);
                }
            });
        }
    }

    /**
     * validate all loaders to avoid duplicated or cycle dependency
     */
    private void checkLoaders() {
        loaders.forEach((name, loader) -> {
            checkLoader(loader);
        });
    }

    private void checkLoader(Loader loader) {
        List<String> dependedLoaders = loader.getSetting().getDependedLoader();
        Set<String> checked = new HashSet<>();
        dependedLoaders.forEach(dloaderName -> {
            Loader dloader = loaders.get(dloaderName);
            if (dloader == null) {
                throw new IllegalStateException("loader " + dloaderName + " not exist, but depended by loader " + loader.getName());
            }
            if (checked.contains(dloaderName)) {
                throw new IllegalStateException("depended loader " + dloaderName + " duplicated or cycle dependency checked.");
            }
            checked.add(dloaderName);
            checkLoader(dloader);
        });
    }

    /**
     * reorg all loaders into {@link #independentLoaders}
     */
    private void reorgLoaders() {
        this.loaders.forEach((lname, loader) -> {
            List<String> names = loader.getSetting().getDependedLoader();
            if (names.isEmpty()) {
                this.independentLoaders.put(lname, loader);
            }
        });
    }

    public void finish(Loader loader) {
        if (this.loaders.containsKey(loader.getName())) {
            if (loader.getStatus() == LoaderStatus.INITED) {
                loader.setStatus(LoaderStatus.SUCCESS);
                this.finishedLoaders.put(loader.getName(), loader);
                this.jobExecution.record(loader);
            }
        }
    }

    public void fail(Loader loader) {
        if (this.loaders.containsKey(loader.getName())) {
            loader.setStatus(LoaderStatus.FAILED);
            this.finishedLoaders.put(loader.getName(), loader);
            this.jobExecution.record(loader);
        }
    }

    /**
     *
     * @throws IOException
     */
    protected void initChecklists() throws IOException {
        List<String> checklists = getJobSetting().getChecklists();
        checklists.forEach(checklistSetting -> {
            try {
                CheckList checklist = ChecklistFactory.getInstance().resolveInstance(checklistSetting, this);
                if (this.checklists.containsKey(checklist.getName())) {
                    throw new EtltRuntimeException("duplicated checklist found: " + checklist.getName());
                }
                this.checklists.put(checklist.getName(), checklist);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
    }

    protected void initMapping() {
        getJobSetting().getMappings().forEach((resourceName) -> {
            try {
                Map mapping =
                        reader.read(new File(this.getContextRoot(), resourceName), Map.class);
                this.mapping.putAll(mapping);
            } catch (IOException e) {
                throw new IllegalStateException(e);
            }
        });
    }

    protected void initResources() {
        List<ResourceSetting> resources = this.jobSetting.getResources();
        ResourceFactory resourceFactory = new ResourceFactory();
        resources.forEach((s) -> {
            Object resource = resourceFactory.createResource(s);
            if (this.resourceContainer.containsKey(s.getName()))
                throw new EtltRuntimeException("Duplicated resource name found: " + s.getName());
            this.resourceContainer.put(s.getName(), resource);
        });
    }

    protected void closeResources() {
        //todo
    }


    public Extractor createExtractor(String name) {
        Extractor extractor = this.extractors.get(name);
        if (Objects.isNull(extractor))
            throw new EtltRuntimeException("extractor not found: " + name);
        return extractor.createInstance();
    }

    public Extractor createExtractor(String name, PkRange pkRange) {
        Extractor extractor = this.extractors.get(name);
        if (Objects.isNull(extractor))
            throw new EtltRuntimeException("extractor not found: " + name);
        return extractor.createInstance(pkRange);
    }

    public Extractor getExtractor(String name) {
        Extractor extractor = this.extractors.get(name);
        if (Objects.isNull(extractor))
            throw new EtltRuntimeException("extractor not found: " + name);
        return extractor;
    }

    public List<Extractor> getAllExtractors() {
        List<Extractor> extractors = new ArrayList<>(this.extractors.values());
        return extractors;
    }

    public Loader getLoader(String name) {
        return this.loaders.get(name);
    }

    public List<Loader> getAllLoaders() {
        List<Loader> loaders = new ArrayList<Loader>(this.loaders.values());
        return loaders;
    }

    /**
     * return all unfinished loaders whose dependent loaders are finished and status is finished
     * {@link LoaderStatus#SUCCESS}
     */
    public List<Loader> getUnfinishedLoaders() {
        List<Loader> result = new ArrayList<>();
        /*
         return all independent loaders
         */
        List<Loader> todoLoaders = new ArrayList<>(this.independentLoaders.values());
        todoLoaders.forEach(loader -> {
            if (!this.finishedLoaders.containsValue(loader)) {
                result.add(loader);
            }
        });
        if (!result.isEmpty())
            return result;

        todoLoaders = new ArrayList<>(this.loaders.values());
        todoLoaders.forEach(loader -> {
            if (loader.getStatus() == LoaderStatus.INITED) {
                boolean skip = false;
                if (!this.finishedLoaders.containsValue(loader)) {
                    List<String> dependLoaderNames = loader.getSetting().getDependedLoader();
                    for (String name : dependLoaderNames) {
                        Loader dependLoader = finishedLoaders.get(name);
                        if (dependLoader == null) {
                            skip = true;
                            break;
                        }
                        if (dependLoader.getStatus() == LoaderStatus.FAILED ||
                                dependLoader.getStatus() == LoaderStatus.UN_EXECUTED) {
                            loader.setStatus(LoaderStatus.UN_EXECUTED);
                            this.finishedLoaders.put(loader.getName(), loader);
                            log.info("skip loader " + loader.getName()
                                    + ", because of depended loader " + dependLoader.getName() + " failed.");
                            skip = true;
                            break;
                        }
                    }
                    if (!skip)
                        result.add(loader);
                }
            }
        });
        return result;
    }

    public List<CheckList> getAllChecklists() {
        List<CheckList> checkLists = new ArrayList<>(this.checklists.values());
        return checkLists;
    }

    public CheckList getChecklist(String name) {
        return this.checklists.get(name);
    }

    /**
     * @param name entity.column
     * @return
     */
    @Override
    public Variable getVariable(String name) {
        int index = name.indexOf('.');
        if (index > 0) {
            String entity = name.substring(0, index);
            String field = name.substring(index + 1);
            return Variable.createVariable(name, getValue(entity, field));
        } else {
            Collection<Entity> values = this.entityContainer.values();
            for (Entity entity : values) {
                if (entity.getValue(name) != null) {
                    return Variable.createVariable(name, entity.getValue(name));
                }
            }
        }
        throw new EtltRuntimeException("variable not found: " + name);
    }

    @SuppressWarnings("unchecked")
    public <T> T getResource(String name) {
        Object resource = this.resourceContainer.get(name);
        if (resource == null)
            throw new EtltRuntimeException("resource not found: " + name);
        return (T) resource;
    }

    /**
     * close this context
     */
    public void close() {
        closeResources();
    }


}
